package com.atguigu.gmall.realtime.utils

import java.util

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.apache.parquet.schema.Types.ListBuilder
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.indices.GetIndexRequest
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.cluster.metadata.Metadata.XContentContext
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.search.SearchHit
import org.elasticsearch.search.builder.SearchSourceBuilder

import scala.collection.mutable.ListBuffer

object MyEsUtils {

  def searchByFiled(indexName :String,field :String) : List[String] ={
    //判断索引是否存在
    val getIndexRequest = new GetIndexRequest(indexName)
    val indexNameIsExists: Boolean = esClient.indices().exists(getIndexRequest,RequestOptions.DEFAULT)
    if(!indexNameIsExists){
      return null
    }
    //从es中提取字段mid
    val searchRequest = new SearchRequest(indexName)
    val searchSourceBuilder = new SearchSourceBuilder()
    searchSourceBuilder.fetchSource(field,null).size(10000)
    searchRequest.source(searchSourceBuilder)
    val searchResponse: SearchResponse = esClient.search(searchRequest,RequestOptions.DEFAULT)
    val hits: Array[SearchHit] = searchResponse.getHits.getHits
    val midBuffer: ListBuffer[String] = ListBuffer[String]()
    for (searchHit <- hits) {
      val sourceMap: util.Map[String, AnyRef] = searchHit.getSourceAsMap
      val mid: String = sourceMap.get(field).toString
      midBuffer.append(mid)
    }
    midBuffer.toList
  }

  //批量写入  String docid  any 数据
  def save(indexName : String , dosc : List[(String,Any)]) ={
    val bulkRequest = new BulkRequest(indexName)
    for ((docId,sourceObj) <- dosc){
      val indexRequest = new IndexRequest()
      indexRequest.id(docId)
      indexRequest.source(JSON.toJSONString(sourceObj,new SerializeConfig(true)),XContentType.JSON)
      bulkRequest.add(indexRequest)
    }

    esClient.bulk(bulkRequest,RequestOptions.DEFAULT)
  }


  val esClient : RestHighLevelClient = create()
  def create() : RestHighLevelClient ={
    val host : String = MyPropertiesUtils(Myconfig.ES_HOST)
    val port : Int = MyPropertiesUtils(Myconfig.ES_PORT).toInt
    val clientBuilder: RestClientBuilder = RestClient.builder(new HttpHost(host,port))
    val client = new RestHighLevelClient(clientBuilder)
    client
  }

  def close() ={
    if(esClient!=null){
      esClient.close()
    }
  }
}
